package com.atguigu.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * @author yhm
 * @create 2023-09-18 14:42
 */
public class HBaseDML {

    public static Connection connection = HBaseConnection.getHBaseConnection();


    /**
     * 写入一条cell
     * @param namespace 命名空间
     * @param tableName 表名
     * @param rowKey 主键
     * @param columnFamily 列族
     * @param columnName 列名
     * @param value 值
     * @throws IOException table连接获取异常
     */
    public static void putCell( String namespace,String tableName,String rowKey,String columnFamily,String columnName,String value) throws IOException {
        // 使用单例的连接  不要在单个方法里面额外创建
        // 1. 获取table实例  填写对应的库名加表名
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));


        // 2. 创建写入数据的对象
        Put put = new Put(Bytes.toBytes(rowKey));

        // 3. 添加对应的数据
        put.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName),Bytes.toBytes(value));

        // 4. table调用对应的方法 写入数据
        try {
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("写入的当前数据有问题" + put.toJSON());
        }

        // 5. 关闭table
        table.close();
    }

    /**
     * 读取hbase的一个cell数据
     * @param namespace 命名空间
     * @param tableName 表名
     * @param rowKey 主键
     * @param columnFamily 列族
     * @param columnName 列名
     * @return 第一个value值
     * @throws IOException hbase的table连接获取异常
     */
    public static String getCell(String namespace,String tableName,String rowKey,String columnFamily,String columnName) throws IOException {

        // 1. 获取table实例
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 2. 创建get对象
        Get get = new Get(Bytes.toBytes(rowKey));

        // 3. 添加列的限定
        get.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName));

        // 4. 调用table的方法get
        Result result = null;
        String value = null;
        try {
            result = table.get(get);
            // 如果只调用1个值
            value = new String(result.value(), StandardCharsets.UTF_8);

            for (Cell cell : result.rawCells()) {
                System.out.println(new String(CellUtil.cloneRow(cell)) + ":" + new String(CellUtil.cloneFamily(cell)) + ":" + new String(CellUtil.cloneQualifier(cell)) + ":" + new String(CellUtil.cloneValue(cell)));
            }
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("读取表格数据异常" + get.toJSON());
        }

        // 关闭连接 释放资源
        table.close();

        return value;

    }

    /**
     * 扫描数据
     * @param namespace 命名空间
     * @param tableName 表名
     * @param startRow 开始行 默认包含
     * @param stopRow 结束行 默认不包含
     * @return 扫描的cell
     * @throws IOException 连接异常
     */
    public static List<String> scanRows(String namespace,String tableName,String startRow,String stopRow) throws IOException {
        // 1. 获取table实例
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 2. 创建scan对象
        Scan scan = new Scan();

        // 3. 限定扫描数据的范围
        // startRow默认是包含的
        scan.withStartRow(Bytes.toBytes(startRow));
        // stopRow默认是不包含的
        scan.withStopRow(Bytes.toBytes(stopRow));

        ArrayList<String> strings = new ArrayList<>();

        // 4. 调用table的扫描数据方法
        // scan得到的结果是多行数据
        ResultScanner scanner = null;
        try {
            scanner = table.getScanner(scan);
            // 5. 处理数据
            for (Result result : scanner) {
                // result对应的就是一行数据
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    StringBuilder stringBuilder = new StringBuilder();
                    // cell对应一行数据中的一列(一个版本)
                    stringBuilder.append(new String(CellUtil.cloneRow(cell)))
                            .append(":")
                            .append(new String(CellUtil.cloneFamily(cell)))
                            .append(":")
                            .append(new String(CellUtil.cloneQualifier(cell)))
                            .append(":")
                            .append(new String(CellUtil.cloneValue(cell)));
                    strings.add(stringBuilder.toString());
                    System.out.print(stringBuilder.toString() + "\t");
                }
                System.out.println();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if (scanner !=null){
                scanner.close();
            }

        }
        // 6. 关闭连接 释放资源

        table.close();

        return strings;

    }

    /**
     * 扫描带过滤器
     * @param namespace 命名空间
     * @param tableName 表名
     * @param startRow 开始行 默认包含
     * @param stopRow 结束行 默认不包含
     * @param familyName 过滤的列族名
     * @param columnName 过滤的列名
     * @param value 过滤的值
     * @return 过滤之后的cell
     * @throws IOException 连接异常
     */
    public static List<String> scanRowsWithFilter(String namespace,String tableName,String startRow,String stopRow,String familyName,String columnName,String value) throws IOException {
        // 1. 获取table实例
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 2. 创建scan对象
        Scan scan = new Scan();

        // 3. 限定扫描数据的范围
        // startRow默认是包含的
        scan.withStartRow(Bytes.toBytes(startRow));
        // stopRow默认是不包含的
        scan.withStopRow(Bytes.toBytes(stopRow));

        // 设置多个过滤器 默认需要全部满足  可以设置为满足一个就行
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);

        // ColumnValueFilter 只保留当前过滤列的数据
//        filterList.addFilter(new ColumnValueFilter(Bytes.toBytes(familyName),Bytes.toBytes(columnName), CompareOperator.EQUAL,Bytes.toBytes(value)));

        // SingleColumnValueFilter 保留过滤出的整行数据
        filterList.addFilter(new SingleColumnValueFilter(Bytes.toBytes(familyName),Bytes.toBytes(columnName), CompareOperator.EQUAL,Bytes.toBytes(value)));

        // 添加过滤器
        scan.setFilter(filterList);

        ArrayList<String> strings = new ArrayList<>();

        // 4. 调用table的扫描数据方法
        // scan得到的结果是多行数据
        ResultScanner scanner = null;
        try {
            scanner = table.getScanner(scan);
            // 5. 处理数据
            for (Result result : scanner) {
                // result对应的就是一行数据
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    StringBuilder stringBuilder = new StringBuilder();
                    // cell对应一行数据中的一列(一个版本)
                    stringBuilder.append(new String(CellUtil.cloneRow(cell)))
                            .append(":")
                            .append(new String(CellUtil.cloneFamily(cell)))
                            .append(":")
                            .append(new String(CellUtil.cloneQualifier(cell)))
                            .append(":")
                            .append(new String(CellUtil.cloneValue(cell)));
                    strings.add(stringBuilder.toString());
                    System.out.print(stringBuilder.toString() + "\t");
                }
                System.out.println();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if (scanner !=null){
                scanner.close();
            }

        }
        // 6. 关闭连接 释放资源

        table.close();

        return strings;

    }

    /**
     * 删除所有版本的列值
     * @param namespace 命名空间
     * @param tableName 表名
     * @param rowKey 主键
     * @param columnFamily 列族
     * @param columnName 列名
     * @throws IOException 连接异常
     */
    public static void deleteCells(String namespace,String tableName,String rowKey,String columnFamily,String columnName) throws IOException {
        // 1. 获取table实例
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 2. 创建delete对象
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        // addColumn 只删除1个版本
//        delete.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName));
        // addColumns 删除所有的版本
        delete.addColumns(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName));

        // 3. 调用table的删除方法
        try {
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 4. 关闭连接 释放资源
        table.close();
    }


    public static void main(String[] args) throws IOException {
        System.out.println(connection);

//        putCell("bigdata","student","100","info","age","10");

//        getCell("bigdata","student","01003","info","age");

//        scanRows("bigdata","student","01000","01005");

//        scanRowsWithFilter("bigdata","student","01000","1007","info","name","lisi");


        deleteCells("bigdata","student","01003","info","age");

        connection.close();
    }
}
